package com.cennavi.funcs;

import java.util.HashMap;
import java.util.List;

import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;

import com.cennavi.beans.DevAndLane;
import com.cennavi.beans.DevAndRtic;
import com.cennavi.beans.SensorFlowData;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月19日 上午9:35:44 
    * @version 1.0 
    */

public class MatchRticFunction implements Function<Tuple2<String, String>, String> {

	private static final long serialVersionUID = 1L;
	//key是dev_id，value是检测的车道id列表
	private HashMap<String, List<DevAndLane>> dev_laneMap;
	//key是dev_id，value是对应的rticid
	private HashMap<String,DevAndRtic> dev_rticMap;
	
	//构造函数
	public MatchRticFunction(
			Broadcast<HashMap<String, List<DevAndLane>>> dev_laneBroadMap,
			Broadcast<HashMap<String, DevAndRtic>> dev_rticBroadMap){
		this.dev_laneMap = dev_laneBroadMap.value();
		this.dev_rticMap = dev_rticBroadMap.value();
	}
	
	public String call(Tuple2<String, String> arg0) throws Exception {
		//输入数据（每行）：dev_id, dev_way_id, flow, time
		String data_source = arg0._2;
		//输出数据（每行末尾加上匹配的rticid）
		String data_matchRtic = "";
		String rtic_id = "";
		String dir = "";
		String[] arr_split = arg0._2.split(",");
		SensorFlowData sensorFlowData = new SensorFlowData();
		sensorFlowData.setDev_id(arr_split[0]);
		sensorFlowData.setDev_way_id(arr_split[1]);
		sensorFlowData.setFlow(Integer.parseInt(arr_split[2]));
		sensorFlowData.setCollect_time(arr_split[3]);
		
//		if(sensorFlowData.getDev_id().equals("90")){
//			System.out.println("");
//		}
		
		if(this.dev_rticMap.containsKey(sensorFlowData.getDev_id())){
//			if(this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic_num().equals("3")){
//				System.out.println("");
//			}
			//如果每个设备只对应一个rtic
			if(this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic_num().equals("1")){
				rtic_id = this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic1_id();
				//System.out.println("******rtic id init at 0*******");
			}
			//每个设备对应rtic数量大于1
			else{
				//设备和车道对应关系表中能找到dev_id
				//1 找到这个dev_id, dev_way_id对应的方向
				//2 根据dev_id, dir找到对应的rtic
				
				if(this.dev_laneMap.containsKey(sensorFlowData.getDev_id() + "E2W")){
					for(int i = 0;i < this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").size();i ++){
						//if(null != this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i)){
						if(this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum1().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum2().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum3().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum4().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum5().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum6().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum7().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum8().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum9().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getLaneNum10().equals(sensorFlowData.getDev_way_id())){
							
							dir = this.dev_laneMap.get(sensorFlowData.getDev_id() + "E2W").get(i).getDir();
							break;
						}
						//}
					}
					
				}
				if(this.dev_laneMap.containsKey(sensorFlowData.getDev_id() + "W2E")){
					for(int i = 0;i < this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").size();i ++){
						//if(null != this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i)){
						if(this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum1().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum2().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum3().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum4().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum5().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum6().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum7().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum8().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum9().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getLaneNum10().equals(sensorFlowData.getDev_way_id())){
							dir = this.dev_laneMap.get(sensorFlowData.getDev_id() + "W2E").get(i).getDir();
							break;
						}
						//}
					}
					
				}
				if(this.dev_laneMap.containsKey(sensorFlowData.getDev_id() + "S2N")){
					for(int i = 0;i < this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").size();i ++){
						//if(null != this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i)){
						if(this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum1().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum2().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum3().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum4().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum5().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum6().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum7().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum8().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum9().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getLaneNum10().equals(sensorFlowData.getDev_way_id())){
							dir = this.dev_laneMap.get(sensorFlowData.getDev_id() + "S2N").get(i).getDir();
							break;
						}
						//}
					}
					
				}
				if(this.dev_laneMap.containsKey(sensorFlowData.getDev_id() + "N2S")){
					for(int i = 0;i < this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").size();i ++){
						//if(null != this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i)){
						if(this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum1().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum2().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum3().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum4().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum5().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum6().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum7().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum8().equals(sensorFlowData.getDev_way_id())
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum9().equals(sensorFlowData.getDev_way_id())	
						|| this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getLaneNum10().equals(sensorFlowData.getDev_way_id())){
							dir = this.dev_laneMap.get(sensorFlowData.getDev_id() + "N2S").get(i).getDir();
							break;
						}
						//}
					}
				}
				if(!dir.equals("")){
					if(this.dev_rticMap.get(sensorFlowData.getDev_id()).getDir1().equals(dir)){
						rtic_id = this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic1_id();
						//System.out.println("******rtic id init at 1*******");
					}
					if(this.dev_rticMap.get(sensorFlowData.getDev_id()).getDir2().equals(dir)){
						rtic_id = this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic2_id();
						//System.out.println("******rtic id init at 2*******");
					}
					if(this.dev_rticMap.get(sensorFlowData.getDev_id()).getDir3().equals(dir)){
						rtic_id = this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic3_id();
						//System.out.println("******rtic id init at 3*******");
					}
					if(this.dev_rticMap.get(sensorFlowData.getDev_id()).getDir4().equals(dir)){
						rtic_id = this.dev_rticMap.get(sensorFlowData.getDev_id()).getRtic4_id();
						//System.out.println("******rtic id init at 4*******");
					}
				}else{
					rtic_id = "NaN";
				}
			}
		}else{
			rtic_id = "NaN";
		}
		if(rtic_id.equals("")){
			//System.out.println(sensorFlowData.getDev_id() + "<------>" + sensorFlowData.getDev_way_id());
			rtic_id = "NaN";
		}
		data_matchRtic += data_source + "," + rtic_id;
		return data_matchRtic;
	}

}
